#include "threadpool.h"

const int MAX_TASK_SIZE = 1024;
const int MAX_THREAD_SIZE = 8;
const int MAX_THREAD_IDLE_TIME = 10;

/*------------------信号量成员函数-----------------*/
Semaphore::Semaphore(int count)
	:count_(count)
{

}

void Semaphore::wait()
{
	std::unique_lock<std::mutex> lock(mtx_);
	cond_.wait(lock, [&]()->bool {return count_ > 0; });
	count_--;
}

void Semaphore::post()
{
	std::unique_lock<std::mutex> lock(mtx_);
	count_++;
	cond_.notify_all();
}

/*------------------task类函数-------------------------*/

//初始化静态变量
int Task::autoincrement_ = 0;

Task::Task()
	:result_(nullptr)
	,taskId_(++autoincrement_)
{

}

void Task::exec()
{
	if (result_ != nullptr)
	{
		result_->setVal(run());
	}

}

void Task::setResult(Result* res)
{
	result_ = res;
}

int Task::getTaskId()
{
	return taskId_;
}

/*---------------Result类-----------------------------*/
Result::Result(std::shared_ptr<Task> task, bool isValid)
	:task_(task)
	,isValid_(isValid)
{
	task_->setResult(this);
}

//用户调用
Any Result::get()
{
	if (!isValid_)
	{
		return "";
	}
	//任务如果没有执行完，这里会阻塞
	sem_.wait();
	return std::move(any_);
}

//任务线程设置
void Result::setVal(Any any)
{
	any_ = std::move(any);
	sem_.post();
	
}


/*-------------线程池类的成员函数-----------------------*/
ThreadPool::ThreadPool()
	:initThreadSize_(4)
	,taskSize_(0)
	,currentThreadSize_(0)
	,idleThreadSize_(0)
	,maxThreadSize_(MAX_THREAD_SIZE)
	,maxTaskSize_(MAX_TASK_SIZE)
	,poolMode_(PoolMode::MODE_FIXED)
	,isPoolRunning_(false)
{

}

ThreadPool::~ThreadPool()
{
	std::cout << "开始析构" << std::endl;
	isPoolRunning_ = false;
	//等待线程池里面所有的线程返回
	std::unique_lock<std::mutex> lock(exitMtx_);
	notEmpty.notify_all();
	std::cout << "~ThreadPool after notify_all" << std::endl;
	exitCond_.wait(lock, [&]()->bool {return currentThreadSize_ == 0; });
	std::cout << "ThreadPool 析构	" << std::endl;
}

void ThreadPool::setMode(PoolMode mode)
{
	if (checkRunningState())
		return;
	poolMode_ = mode;
}

void ThreadPool::setMaxTaskSize(int maxSize)
{
	if (checkRunningState())
		return;
	maxTaskSize_ = maxSize;
}

void ThreadPool::setMaxThreadSize(int size)
{
	if (checkRunningState())
		return;
	if (poolMode_ == PoolMode::MODE_CACHED)
	{
		maxThreadSize_ = size;
	}
}

Result ThreadPool::submitTask(std::shared_ptr<Task> spTask)
{
	//建立unique_lock并且锁上
	std::unique_lock<std::mutex> lock(taskQueMtx);

	/*等待任务队列有空闲wait一般方法
	while (taskQue.size() == maxTaskSize)
	{
		notFull.wait(lock);
	}
	*/

	/*wait在c++11提供的lambda表达式方法
	notFull.wait(lock, [&]()->bool {return taskQue.size() < maxTaskSize; });
	*/

	//使用wait_for方法可以兼顾用户超时等待的情况，返回false表示等待约定时间条件依然没有成立
	if (!notFull.wait_for(lock, std::chrono::seconds(1), [&]()->bool {return taskQue_.size() < static_cast<size_t>(maxTaskSize_); }))
	{
		//表示notFull等待1秒后，条件依然没有满足
		std::cerr << "task queue is full, submit task fail." << std::endl;
		return Result(spTask, false);
	}
	
	//如果有空余，把任务放入任务队列中
	taskQue_.emplace(spTask);
	taskSize_++;
	std::cout << "added task id = " << spTask->getTaskId() << std::endl;
	notEmpty.notify_all();

	//cached模式任务处理比较紧急 场景：小而快的任务
	if (poolMode_ == PoolMode::MODE_CACHED && taskSize_ > idleThreadSize_ && currentThreadSize_ < maxThreadSize_)
	{
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this,std::placeholders::_1));
		int threadId = ptr->getId();
		threads.emplace(threadId, std::move(ptr));
		//启动线程
		threads[threadId]->start();
		std::cout << "动态创建线程threadID=" << threadId << std::endl;
		currentThreadSize_++;
		idleThreadSize_++;
	}

	//此处设置了spTask->result_的值
	return Result(spTask);
}

void ThreadPool::start(int initThreadSize)
{
	//记录初始线程个数
	initThreadSize_ = initThreadSize;
	currentThreadSize_ = initThreadSize;
	//设置线程池的运行状态
	isPoolRunning_ = true;
	//创建线程对象，emplace_back创建比push_back效率高
	for (size_t i = 0; i < initThreadSize_; i++)
	{
		//创建thread线程对象的时候，把线程函数给到thread线程对象
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this,std::placeholders::_1));
		int threadId = ptr->getId();
		std::cout << "创建线程threadId=" << threadId << std::endl;
		threads.emplace(threadId, std::move(ptr));
	}

	//启动所有线程 
	for (size_t i = 0; i < initThreadSize_; i++)
	{
		threads[i]->start();
		idleThreadSize_++;
	}
}

//定义线程函数
void ThreadPool:: threadFunc(int threadId)
{
	auto lastTime = std::chrono::high_resolution_clock().now();
	while (1)
	{

		std::shared_ptr<Task> task;
		{
			std::unique_lock<std::mutex> lock(taskQueMtx);

			//cached模式下，有可能已经创建了很多的线程，但是空闲时间超过60s，应该把多余的线程回收
				//每秒钟返回一次
			//锁+双重判断

			std::cout << "任务队列个数 = " << taskQue_.size() << "当前线程为"<< threadId<<std::endl;
			while (taskQue_.size() == 0)
			{
				//线程池结束回收线程资源
				std::cout << "isPoolRunning = " << checkRunningState() << std::endl;
				if (!checkRunningState())
				{

					threads.erase(threadId);
					std::cout << "erase threadid= " << threadId << "  tid= " << std::this_thread::get_id() << std::endl;
					currentThreadSize_--;
					if (0 == currentThreadSize_)
					{
						std::cout << "exitCond.notify threadid= " << threadId << "  tid= " << std::this_thread::get_id() << std::endl;
						exitCond_.notify_all();
					}
					return;
				}
				if (poolMode_ == PoolMode::MODE_CACHED)
				{

					if (std::cv_status::timeout == notEmpty.wait_for(lock, std::chrono::seconds(1)))
					{
						auto now = std::chrono::high_resolution_clock().now();
						auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
						//大于60秒回收线程
						if (currentThreadSize_ > initThreadSize_ && MAX_THREAD_IDLE_TIME <= dur.count())
						{
							//开始回收线程
							//记录线程相关数据
							//把线程对象从线程池里丢弃
							threads.erase(threadId);
							currentThreadSize_--;
							idleThreadSize_--;
							std::cout << "erase threadid=" << threadId << "  tid=" << std::this_thread::get_id() << std::endl;
							return;
						}
					}
				}
				else
				{
					std::cout << "before notEmpty.wait  " <<"threadid="<< threadId << std::endl;
					notEmpty.wait(lock);
					std::cout << "after notEmpty.wait  " << "threadid=" << threadId << std::endl;
				}
			}
			idleThreadSize_--;

			task = taskQue_.front();
			taskQue_.pop();
			taskSize_--;
			//如果此时还有任务，继续通知其他线程执行
		}
		if (taskQue_.size() > 0)
		{
			notEmpty.notify_all();
		}
		notFull.notify_all();


		if (task != nullptr)
		{
			task->exec();

		}
		idleThreadSize_++;
		lastTime = std::chrono::high_resolution_clock().now();
	}

}

bool ThreadPool::checkRunningState() const
{
	return isPoolRunning_;
}


/*------------------Thread----------------------*/

int Thread::generateId_ = 0;//初始化静态变量

Thread::Thread(ThreadFunc function)
	:func(function)
	,threadId_(generateId_++)
{

}

Thread::~Thread()
{

}

void Thread::start()
{
	//创建一个线程来执行一个线程函数
	std::thread t(func,threadId_); //c++11来说线程对象t 和线程函数func
	t.detach(); //设置线程的分离属性
}

int Thread::getId() const
{
	return threadId_;
}
